/**
 * %datax%
 * %v1.0%
 */
package com.alibaba.datax.elasticresource.watcher;

import org.apache.logging.log4j.util.Strings;

import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.core.config.Key;
import com.alibaba.datax.core.statistics.SnapshotStatistics;
import com.alibaba.datax.core.statistics.SummaryStatistics;
import com.alibaba.datax.core.util.container.CoreConstant;
import com.leehom.arch.base.redis.client.RedisClient;
import com.leehom.arch.datax.report.core.repository.StatisticsReportExporter;
import com.leehom.arch.datax.report.core.repository.StatisticsReportRepository;
import com.leehom.arch.datax.report.core.repository.StatsticsReportNopRepository;
import com.leehom.arch.datax.report.repository.elasticsearch.StatisticsReportElasticsearchExporter;
import com.leehom.arch.datax.report.repository.log.StatisticsReportLogRepository;
import com.leehom.arch.datax.report.repository.redis.StatisticsReportRedisRepository;
import com.leehom.arch.onesearch.index.DocumentService;

/**
 * @类名: ReporterFactory
 * @说明: 报告器工厂
 *                   作业通讯器      任务组通讯器      watcher
 *        单机           N/A*           N/A*            无
 *        分布式                       redis       elasticsearch
 *        
 *        *单机，JobContainer带统计打印
 *        **分布式，作业通讯器可忽略，功能转移到watcher
 *        
 *
 * @author   leehom
 * @Date	 2022年10月31日 下午5:22:08
 * 修改记录：
 *
 * @see 	 
 */
public class WatcherReporterFactory {

	//
	public static StatisticsReportExporter<SummaryStatistics, Long> getStatisticsReportExporter(Configuration conf) {
		Configuration summaryConf = conf.getConfiguration(CoreConstant.DATAX_CORE_STATISTICS_SUMMARY_REPORT);
		String type = Key.summaryReportType(summaryConf);
		if("log".equals(type))
			return new StatisticsReportLogRepository<SnapshotStatistics, SummaryStatistics, Long>();
		else if("elasticsearch".equals(type)) {
			// 
			DocumentService docService = WatcherReporterBeanConfig.buildElasticsearchClient(conf);
			//
			Configuration localConf = conf.getConfiguration(CoreConstant.DATAX_CORE_STATISTICS_SUMMARY_REPORT_ELASTICSEARCH);
			String index = Key.elasticsaerchIndex(localConf);
			//
			StatisticsReportElasticsearchExporter<SummaryStatistics, Long> rr = 
						new StatisticsReportElasticsearchExporter<>(index, docService);
			return rr;
		}
		// 最后没有适配，返回nop
		return new StatsticsReportNopRepository<SnapshotStatistics, SummaryStatistics, Long>();
	}
	
	public static StatisticsReportRepository<SnapshotStatistics, SummaryStatistics, Long> getSnapshotReportRepository(Configuration conf) {
		Configuration localConf = conf.getConfiguration(CoreConstant.DATAX_CORE_STATISTICS_SNAPSHOT_REPORT);
		String type = Key.snapshotReportType(localConf);
		if("log".equals(type))
			return new StatisticsReportLogRepository<SnapshotStatistics, SummaryStatistics, Long>();
		else if("redis".equals(type)) { //
			Configuration redisConf = conf.getConfiguration(CoreConstant.DATAX_CORE_STATISTICS_SNAPSHOT_REPORT_REDIS);
			String keyPrefix = Key.redisClientKeyPrefix(redisConf);
			if(Strings.isBlank(keyPrefix))
				keyPrefix = "";
			long expiredTime = Key.redisClientExpireTime(redisConf);
			//
			RedisClient redisClient = getRedisClient(conf);
			//
			return new StatisticsReportRedisRepository<SnapshotStatistics, SummaryStatistics, Long>(keyPrefix, expiredTime, redisClient);
		}
		// 最后没有适配，返回nop
		return new StatsticsReportNopRepository<SnapshotStatistics, SummaryStatistics, Long>();
		
	}
	
	private static volatile RedisClient redisClient;
	public static RedisClient getRedisClient(Configuration conf) {
    	if(redisClient==null) {
    		synchronized (RedisClient.class) {
    			if(redisClient==null) {
					Configuration redisConf = conf.getConfiguration(CoreConstant.DATAX_CORE_STATISTICS_SNAPSHOT_REPORT_REDIS);
					// redis类型
					String redisType = Key.redisClientType(redisConf);
					//
					if("sentinel".equals(redisType))
						redisClient = WatcherReporterBeanConfig.sentinelRedis(conf);
					else if("cluster".equals(redisType))
						redisClient = WatcherReporterBeanConfig.clusterRedis(conf);
					else
						redisClient = WatcherReporterBeanConfig.standaloneRedis(conf);
    			}
    		}
    	}
		return redisClient;
	}
	
}
